package io.gitee.skyarthur1987.leporidae.delayqueue.core;

import io.gitee.skyarthur1987.leporidae.delayqueue.api.DelayQueue;
import io.gitee.skyarthur1987.leporidae.delayqueue.dto.BatchDelayQueueTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.function.Consumer;

/**
 * 延迟队列消费者
 * Created by skyarthur on 2023/1/27
 */
@Slf4j
public class DelayQueueConsumer implements Consumer<BatchDelayQueueTask> {
    private final DelayQueue<String> delayQueue;
    private final ApplicationEventPublisher applicationEventPublisher;

    public DelayQueueConsumer(DelayQueue<String> delayQueue, ApplicationEventPublisher applicationEventPublisher) {
        this.delayQueue = delayQueue;
        this.applicationEventPublisher = applicationEventPublisher;
    }

    @Override
    public void accept(BatchDelayQueueTask batchDelayQueueTask) {
        List<String> tasks = delayQueue.poll(batchDelayQueueTask.getTopic(),
                batchDelayQueueTask.getExpireTime());
        if (!CollectionUtils.isEmpty(tasks)) {
            log.info("Publish poll event, topic: {}, expireTime: {}, task size: {}",
                    batchDelayQueueTask.getTopic(), batchDelayQueueTask.getExpireTime(),
                    tasks.size());
            batchDelayQueueTask.setTasks(tasks);
            DelayQueuePollEvent pollEvent = new DelayQueuePollEvent(this, batchDelayQueueTask);
            applicationEventPublisher.publishEvent(pollEvent);
        }
    }
}
